package cm.kafka.configuration;

import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author liuwei
 * @date 2019-01-07
 */
@Configuration
public class ElasticsearchClientConfig {


    @Value("${elasticsearch.hosts}")
    private String hosts;

    @Value("${elasticsearch.ports}")
    private String ports;

    private static final int CONNECT_TIME_OUT = 1000;
    private static final int SOCKET_TIME_OUT = 30000;
    private static final int CONNECTION_REQUEST_TIME_OUT = 5000;

    private static final int MAX_CONNECT_NUM = 100;
    private static final int MAX_CONNECT_PER_ROUTE = 100;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        HttpHost[] httpHosts;
        String[] host = hosts.split(",");
        String[] port = ports.split(",");
        httpHosts = new HttpHost[host.length];
        for (int i = 0; i < host.length; i++) {
            HttpHost httpHost = new HttpHost(host[i], Integer.valueOf(port[i]));
            httpHosts[i] = httpHost;
        }
        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);

        // 设置关于异步httpclient的连接延时配置
        restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {

            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                requestConfigBuilder.setSocketTimeout(SOCKET_TIME_OUT);
                requestConfigBuilder.setConnectTimeout(CONNECT_TIME_OUT);
                requestConfigBuilder.setConnectionRequestTimeout(CONNECTION_REQUEST_TIME_OUT);
                return requestConfigBuilder;
            }
        });
        // 关于异步httpclient的连接数配置
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {

            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                //线程设置
                httpClientBuilder.setMaxConnTotal(MAX_CONNECT_NUM);
                httpClientBuilder.setMaxConnPerRoute(MAX_CONNECT_PER_ROUTE);
                httpClientBuilder.setMaxConnTotal(100);
                httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(100).build());
                return httpClientBuilder;
            }
        });

        RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(httpHosts));
        return client;
    }

    @Bean
    public RestClient restClient() {
        HttpHost[] httpHosts;
        String[] host = hosts.split(",");
        String[] port = ports.split(",");
        httpHosts = new HttpHost[host.length];
        for (int i = 0; i < host.length; i++) {
            HttpHost httpHost = new HttpHost(host[i], Integer.valueOf(port[i]));
            httpHosts[i] = httpHost;
        }
        RestClientBuilder builder = RestClient.builder(httpHosts).setRequestConfigCallback(
                new RestClientBuilder.RequestConfigCallback() {
                    @Override
                    public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                        requestConfigBuilder.setConnectTimeout(50000);
                        requestConfigBuilder.setSocketTimeout(400000);
                        requestConfigBuilder.setConnectionRequestTimeout(10000);
                        return requestConfigBuilder;
                    }
                });

        return builder.build();
    }


}